package com.atguigu.edu.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class HBaseUtil {
    //获取hbase链接
    public static Connection getHbaseConnection(){

        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", "hadoop102");
            conf.set("hbase.zookeeper.property.clientPort", "2181");

            return  ConnectionFactory.createConnection(conf);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    //关闭Hbase链接
    public static void closeHBaseConnection(Connection hbaseConn){
        if (hbaseConn!=null && !hbaseConn.isClosed()){
            try {
                hbaseConn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //在Hbase中建表
    public static void createHBaseTable(Connection hbaseConn,
                                        String nameSpace,
                                        String table,
                                        String ... families){
        //判断是否传递了列族
        if (families.length<1){
            System.out.println("加标必须提供列族");
            return;
        }
        try (Admin admin = hbaseConn.getAdmin()){
            //判断表是否存在
            TableName tableName = TableName.valueOf(nameSpace, table);
            if(admin.tableExists(tableName)){
                System.out.println("要删除的"+nameSpace+"下的"+tableName+"表已存在");
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
            for (String family : families) {
                tableDescriptorBuilder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build());

            }


            admin.createTable(tableDescriptorBuilder.build());
        }catch (IOException e){
            throw new RuntimeException(e);
        }
    }


    //Hbase中删除表
    public static void dropHBaseTable(Connection hbaseConn,
                                      String nameSpace,
                                      String table)  {
        //Admin(DDL)使用完毕需要释放资源 可以try catch finally 也可以放到小括号中，用完就直接释放资源了
        //Table(DML)
        try (Admin admin = hbaseConn.getAdmin()){
            TableName tableName = TableName.valueOf(nameSpace, table);
            if(!admin.tableExists(tableName)){
                System.out.println("要删除的"+nameSpace+"下的"+tableName+"表不存在");
                return;
            }
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("从Hbase删除了"+nameSpace+"下的"+tableName+"表已删除");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
    //从HBase中删除记录
    public static void delRow(Connection connection, String nameSpace, String tableName, String rowKey){

        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = connection.getTable(tableNameObj)){
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            System.out.println("从"+nameSpace+"下的"+tableName+"表中删除"+rowKey+"数据成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //从Hbase表中put记录
    public static void putRow(Connection connection, String nameSpace, String tableName, String rowKey, String family, JSONObject jsonObject){

        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = connection.getTable(tableNameObj)){
            Put put = new Put(Bytes.toBytes(rowKey));
            Set<String> columns = jsonObject.keySet();
            for (String column : columns) {
                String value  = jsonObject.getString(column);
                if(StringUtils.isNoneEmpty(value)){
                    put.addColumn(Bytes.toBytes(family),Bytes.toBytes(column),Bytes.toBytes(value));
                }
            }
            table.put(put);
            System.out.println("从"+nameSpace+"下的"+tableName+"表中put"+rowKey+"数据成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //根据rowKey到Hbase中获取表信息
    public  static <T>T readRow(Connection hbaseConn,String nameSpace,String tableName,String rowKey,Class<T> clz,boolean...isUnderLineToCamel){
        boolean defaultIsUToC = false;  // 默认不执行下划线转驼峰

        if (isUnderLineToCamel.length>0){
            defaultIsUToC=isUnderLineToCamel[0];
        }

        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)){
            Result result = table.get(new Get(Bytes.toBytes(rowKey)));
            List<Cell> cells = result.listCells();
            if(cells!=null&&cells.size()>0){
                //定义一个泛型类，用于封装查询出来的维度数据
                T obj  = clz.newInstance();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    if(defaultIsUToC){
                        columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL,columnName);
                    }
                    BeanUtils.setProperty(obj,columnName,columnValue);

                }
                return obj;
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    /**
     * 获取异步Hbase客户端连接
     * @return
     */
    public static AsyncConnection getAsyncConnection(){
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum","hadoop102");
        AsyncConnection asyncConnection = null;
        try {
            asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
        }
        catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }
        return asyncConnection;
    }

    /**
     * 关闭hbase异步客户端连接
     * @param connection
     */
    public static void closeAsyncConnection(AsyncConnection connection){
        if (connection!=null&& !connection.isClosed()){
            try {
                connection.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 异步从hbase中获取维度信息
     * @param conn
     * @param namespace
     * @param tableName
     * @param rowKey
     * @return
     */
    public static JSONObject readDimAsync(AsyncConnection conn,String namespace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        AsyncTable<AdvancedScanResultConsumer> table = conn.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result result = table.get(get).get();
            List<Cell> cells = result.listCells();
            if (cells!=null&&cells.size()>0){
                JSONObject jsonObject = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObject.put(columnName,columnValue);
                }
                return jsonObject;
            }
        } catch (RuntimeException | InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        return null;
    }
    //以异步的方式从Hbase中根据rowkey获取维度对象
    public static JSONObject getRowAsync(AsyncConnection asyncConn,String namespace, String tableName, String rowKey ){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConn.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }
}